#include "network_libevent.h"
#include <event2/thread.h>
#include <assert.h>
#include <string.h>

#ifdef WIN32
#   include <winsock2.h>
#else
#   include <sys/socket.h>
#   include <netinet/in.h>
#   include <arpa/inet.h>
#endif

namespace engine
{
	static bool isIPV4(const char* host)
	{
		const size_t p = strlen(host) - 1;
		return (host[0] >= '0' && host[0] <= '9') && (host[p] >= '0' && host[p] <= '9');
	}

	static uint32 domain2IPV4(const char* host)
	{
		struct hostent* hent = gethostbyname(host);

		if (hent == 0 || hent->h_addr == 0)
			return INADDR_NONE;

		return *reinterpret_cast<uint32*>(hent->h_addr);
	}

	static uint32 fromHost(const char *host)
	{
		if (host == NULL || host[0] == 0)
			return INADDR_ANY;
		else if (!isIPV4(host))
			return domain2IPV4(host);
		else
		{
			const uint32 ipv4 = inet_addr(host);
			if (ipv4 != INADDR_NONE)
				return ipv4;
			else
				return domain2IPV4(host);
		}
	}

	//////////////////////////////////////////////////////////////////////////

	LibEvent::LibEvent() : mMaxConnections(0), mStopThread(true), mEnterLoop(false),
		mStartSucc(false), mEventBase(NULL), mListener(NULL), mConnectionCount(0)
	{
		mBuffPoolFF.setPageSize(0xFF * 512);
		mBuffPoolFFF.setPageSize(0xFFF * 256);
		mBuffPoolFFFF.setPageSize(0xFFFF * 128);
	}

	void* LibEvent::entry()
	{
		event_config* config = event_config_new();

#ifdef WIN32

		evthread_use_windows_threads();

		if (mMaxConnections != 0)
		{
			event_config_set_flag(config, EVENT_BASE_FLAG_STARTUP_IOCP);

			SYSTEM_INFO si;
			GetSystemInfo(&si);

			event_config_set_num_cpus_hint(config, si.dwNumberOfProcessors);
		}

#else
		evthread_use_pthreads();

		event_config_set_flag(config, EVENT_BASE_FLAG_EPOLL_USE_CHANGELIST);
#endif

		mEventBase = event_base_new_with_config(config);

		event_config_free(config);

		if (mEventBase && mMaxConnections != 0)
		{
			struct sockaddr_in sin;

			memset(&sin, 0, sizeof(struct sockaddr_in));

			sin.sin_family = AF_INET;
			sin.sin_addr.s_addr = mAddress.host;
			sin.sin_port = htons(mAddress.port);

			unsigned flag = LEV_OPT_REUSEABLE | LEV_OPT_CLOSE_ON_FREE | LEV_OPT_THREADSAFE;

			mListener = evconnlistener_new_bind(mEventBase,
				&LibEvent::EVCONNLISTENER_CALLBACK,
				this,
				flag,
				-1,
				(struct sockaddr*)&sin,
				sizeof(struct sockaddr_in));
		}

		mStartSucc = mEventBase && ((mMaxConnections == 0) || (mListener != NULL));

		mEnterLoop = true;

		while (!mStopThread)
		{
			processCommands();
			dispatchEvents();
		}

		mEnterLoop = false;

		evconnlistener_free(mListener);

		mListener = NULL;

		event_base_free(mEventBase);

		mEventBase = NULL;

		return 0;
	}

	bool LibEvent::start(const char* host, uint16 port, uint32 maxConnections)
	{
		if (Thread::isAlive())
			return false;

		mStopThread = false;

		mAddress.host = fromHost(host);
		mAddress.port = port;

		mMaxConnections = maxConnections;
		mConnectionCount = 0;

		if (THREAD_NO_ERROR == Thread::create())
		{
			if (THREAD_NO_ERROR == Thread::run())
			{
				while (!mEnterLoop)
				{
					Thread::sleep(1);
				}

				return mStartSucc;
			}
		}

		mStopThread = true;

		return false;
	}

	void LibEvent::stop()
	{
		mStartSucc = false;

		mStopThread = true;

		Thread::waitExit();

		mConnectionPool.clear();

		mCommands.clear();
		mMessages.clear();

		mBuffPoolFF.clear();
		mBuffPoolFFF.clear();
		mBuffPoolFFFF.clear();
	}

	bool LibEvent::connect(const char* host, uint16 port)
	{
		if (mEnterLoop)
		{
			Command* cmd = mCommands.writeLock();

			cmd->data.addr.host = fromHost(host);
			cmd->data.addr.port = port;
			cmd->type = Command::Connect;

			mCommands.writeUnlock();

			return true;
		}

		return false;
	}

	void LibEvent::disconnect(Connection* conn)
	{
		if (mEnterLoop)
		{
			Command* cmd = mCommands.writeLock();

			cmd->data.conn = conn;
			cmd->type = Command::Disconnect;

			mCommands.writeUnlock();
		}
	}

	void LibEvent::send(void* data, uint16 size, Connection* conn)
	{
		if (mEnterLoop)
		{
			uint32 allsize = size + sizeof(uint16);

			uint8* buff = (uint8*)allocBuffer(allsize);

			if (buff)
			{
				memcpy(buff, &size, sizeof(uint16));

				memcpy(buff + sizeof(uint16), data, size);

				Command* cmd = mCommands.writeLock();

				cmd->data.send.conn = conn;
				cmd->data.send.buff = buff;
				cmd->data.send.size = allsize;
				cmd->type = Command::Send;

				mCommands.writeUnlock();
			}
			else
			{
				assert(false);
			}
		}
	}

	void LibEvent::processCommands()
	{
		Command* pcmd = NULL;

		while ((pcmd = mCommands.readLock()))
		{
			Command cmd = *pcmd;

			mCommands.readUnlock();

			if (cmd.type == Command::Connect)
			{
				processConnect(cmd.data.addr);
			}
			else if (cmd.type == Command::Disconnect)
			{
				processDisconnect(cmd.data.conn);
			}
			else if (cmd.type == Command::Send)
			{
				if (cmd.data.send.conn->buffevt)
				{
					if (0 != bufferevent_write(cmd.data.send.conn->buffevt, cmd.data.send.buff, cmd.data.send.size))
					{
						assert(false);
					}

					freeBuffer(cmd.data.send.buff, cmd.data.send.size);
				}
			}
		}
	}

	void LibEvent::dispatchEvents()
	{
		struct timeval timev;
		timev.tv_sec = 0;
		timev.tv_usec = 1000;
		event_base_loopexit(mEventBase, &timev);
		event_base_dispatch(mEventBase);
	}

	void LibEvent::processConnect(const Address& addr)
	{
		if (mConnectionCount > 0)
		{
			pushMessage(Message::ConnectionFailed);

			return;
		}

		bufferevent* buffev = bufferevent_socket_new(
			mEventBase,
			-1,
			BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE | BEV_OPT_DEFER_CALLBACKS);

		if (buffev == NULL)
		{
			pushMessage(Message::ConnectionFailed);

			return;
		}

		struct sockaddr_in sin;

		memset(&sin, 0, sizeof(struct sockaddr_in));

		sin.sin_family = AF_INET;
		sin.sin_addr.s_addr = addr.host;
		sin.sin_port = htons(addr.port);

		if (bufferevent_socket_connect(buffev, (struct sockaddr*)&sin, sizeof(struct sockaddr_in)) != 0)
		{
			bufferevent_free(buffev);

			pushMessage(Message::ConnectionFailed);

			return;
		}

		Connection* conn = allocConnection();

		conn->address = addr;
		conn->buffevt = buffev;
		conn->connected = false;
		conn->libevent = this;
		conn->bind = NULL;
		conn->buff = NULL;
		conn->buffpos = 0;
		conn->buffsize = 0;

		bufferevent_setcb(buffev,
			&LibEvent::BUFFEREVENT_READ_CALLBACK,
			&LibEvent::BUFFEREVENT_WRITE_CALLBACK,
			&LibEvent::BUFFEREVENT_EVENT_CALLBACK,
			conn);

		bufferevent_enable(buffev, EV_READ | EV_WRITE | EV_TIMEOUT);
	}

	void LibEvent::processDisconnect(Connection* conn)
	{
		if (conn->connected)
		{
			mConnectionCount--;

			assert(conn->buffevt);

			bufferevent_free(conn->buffevt);

			conn->buffevt = NULL;
			conn->address.host = -1;
			conn->address.port = -1;
			conn->connected = false;

			pushMessage(Message::ConnectionBreak, conn);
		}
	}

	void LibEvent::EVCONNLISTENER_CALLBACK(struct evconnlistener* listener, evutil_socket_t fd, struct sockaddr* sin, int len, void* ctx)
	{
		LibEvent* libevent = (LibEvent*)(ctx);

		libevent->processAccept(listener, fd, sin, len);
	}

	void LibEvent::BUFFEREVENT_READ_CALLBACK(struct bufferevent* buffevt, void* ctx)
	{
		Connection* conn = (Connection*)(ctx);

		conn->libevent->processRead(buffevt, ctx);
	}

	void LibEvent::BUFFEREVENT_WRITE_CALLBACK(struct bufferevent* buffevt, void* ctx)
	{
	}

	void LibEvent::BUFFEREVENT_EVENT_CALLBACK(struct bufferevent* buffevt, short what, void* ctx)
	{
		Connection* conn = (Connection*)(ctx);

		conn->libevent->processEvent(buffevt, what, ctx);
	}

	void LibEvent::processAccept(struct evconnlistener* listener, evutil_socket_t fd, struct sockaddr* sin, int len)
	{
		if (listener != mListener)
			return;

		if (mConnectionCount >= mMaxConnections)
		{
			evutil_closesocket(fd);
			return;
		}

		struct bufferevent* buffevt = bufferevent_socket_new(
			mEventBase, fd, BEV_OPT_CLOSE_ON_FREE | BEV_OPT_THREADSAFE | BEV_OPT_DEFER_CALLBACKS);

		if (!buffevt)
			return;

		Connection* conn = allocConnection();

		mConnectionCount++;

		struct sockaddr_in* rsin = (sockaddr_in*)(sin);
		conn->address.host = rsin->sin_addr.s_addr;
		conn->address.port = rsin->sin_port;
		conn->buffevt = buffevt;
		conn->bind = NULL;
		conn->connected = true;
		conn->libevent = this;
		conn->buff = NULL;
		conn->buffpos = 0;
		conn->buffsize = 0;

		bufferevent_setcb(buffevt,
			&LibEvent::BUFFEREVENT_READ_CALLBACK,
			&LibEvent::BUFFEREVENT_WRITE_CALLBACK,
			&LibEvent::BUFFEREVENT_EVENT_CALLBACK,
			conn);

		bufferevent_enable(buffevt, EV_READ | EV_WRITE | EV_TIMEOUT);

		pushMessage(Message::ConnectionAccepted, conn);
	}

	void LibEvent::processRead(struct bufferevent* buffevt, void* ctx)
	{
		Connection* conn = (Connection*)(ctx);

		assert(conn->buffevt == buffevt);

		struct evbuffer* buf = bufferevent_get_input(buffevt);

		size_t len = evbuffer_get_length(buf);

		while (len > 0)
		{
			if (conn->buff == NULL)
			{
				if (len >= sizeof(uint16))
				{
					uint16 msglen = 0;
					void* msgbuff = NULL;

					size_t readBytes = bufferevent_read(conn->buffevt, &msglen, sizeof(uint16));

					assert(readBytes == sizeof(uint16));

					if (msglen > 0)
					{
						msgbuff = allocBuffer(msglen);
					}

					conn->buff = msgbuff;
					conn->buffsize = msglen;
					conn->buffpos = 0;
				}
				else
				{
					break;
				}
			}
			else
			{
				uint16 toread = conn->buffsize - conn->buffpos;

				if (toread > 0)
				{
					uint8* currpos = (uint8*)conn->buff + conn->buffpos;

					size_t readed = bufferevent_read(conn->buffevt, currpos, toread);

					conn->buffpos += (uint16)readed;
				}

				if (conn->buffpos == conn->buffsize)
				{
					pushMessage(Message::ConnectionMessage, conn, conn->buff, conn->buffsize);

					conn->buff = NULL;
					conn->buffpos = 0;
					conn->buffsize = 0;
				}
			}

			buf = bufferevent_get_input(buffevt);
			len = evbuffer_get_length(buf);
		}
	}

	void LibEvent::processEvent(struct bufferevent* buffevt, short what, void* ctx)
	{
		Connection* conn = (Connection*)(ctx);

		assert(conn->buffevt == buffevt);

		if ((what & BEV_EVENT_CONNECTED) != 0)
		{
			mConnectionCount++;
			conn->connected = true;

			pushMessage(Message::ConnectionSuccess, conn);
		}
		else if ((what & BEV_EVENT_ERROR) != 0 || (what & BEV_EVENT_EOF) != 0 || (what & BEV_EVENT_TIMEOUT) != 0 || (what & BEV_EVENT_READING) != 0	|| (what & BEV_EVENT_WRITING) != 0)
		{
			if (conn->connected)
			{
				mConnectionCount--;

				pushMessage(Message::ConnectionBreak, conn);
			}
			else
			{
				pushMessage(Message::ConnectionFailed, conn);
			}

			if (conn->buff)
			{
				freeBuffer(conn->buff, conn->buffsize);
			}

			bufferevent_free(buffevt);

			conn->address.host = -1;
			conn->address.port = -1;
			conn->buffevt = NULL;
			conn->buff = NULL;
			conn->buffpos = 0;
			conn->buffsize = 0;
			conn->connected = false;
		}
	}

	Message* LibEvent::lockMessage()
	{
		return mMessages.readLock();
	}

	void LibEvent::unlockMessage()
	{
		mMessages.readUnlock();
	}

	void LibEvent::pushMessage(Message::Type type, Connection* conn, void* buff, uint16 size)
	{
		Message* msg = mMessages.writeLock();

		msg->type = type;
		msg->conn = conn;
		msg->buff = buff;
		msg->size = size;

		mMessages.writeUnlock();
	}

	void* LibEvent::allocBuffer(uint32 size)
	{
		CriticalSectionLocker lock(mBuffPoolCS);

		void* buff = NULL;

		if (size == 0)
		{
			buff = NULL;
			assert(false);
		}
		else if (size <= 0xFF)
		{
			buff = mBuffPoolFF.alloc();
		}
		else if (size <= 0xFFF)
		{
			buff = mBuffPoolFFF.alloc();
		}
		else if (size <= 0xFFFF)
		{
			buff = mBuffPoolFFFF.alloc();
		}
		else
		{
			buff = NULL;
			assert(false);
		}

		return buff;
	}

	void LibEvent::freeBuffer(void* buff, uint32 size)
	{
		CriticalSectionLocker lock(mBuffPoolCS);

		if (size == 0)
		{
			assert(false);
		}
		else if (size <= 0xFF)
		{
			mBuffPoolFF.release((BufferFF*)buff);
		}
		else if (size <= 0xFFF)
		{
			mBuffPoolFFF.release((BufferFFF*)buff);
		}
		else if (size <= 0xFFFF)
		{
			mBuffPoolFFFF.release((BufferFFFF*)buff);
		}
		else
		{
			assert(false);
		}
	}

	Connection* LibEvent::allocConnection()
	{
		CriticalSectionLocker lock(mConnectionPoolCS);
		return mConnectionPool.alloc();
	}

	void LibEvent::freeConnection(Connection* conn)
	{
		CriticalSectionLocker lock(mConnectionPoolCS);
		mConnectionPool.release(conn);
	}

	void LibEvent::bindData(Connection* conn, void* data)
	{
		conn->bind = data;
	}

	void* LibEvent::bindData(Connection* conn)
	{
		return conn->bind;
	}

};
